package zhuzx.obsolete;

import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.spi.AbstractInterruptibleChannel;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;
import java.util.Vector;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.jcraft.jsch.Channel;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.Session;
import org.zhuzx.util.ConcurrentUtil;
import org.zhuzx.util.GeneralUtil;
import org.zhuzx.util.WebUtil;

public class RemoteExtractor {
	
	private static BlockingQueue<TempFile> container;
	private static final AtomicInteger flag = new AtomicInteger();

	public static Session sshLogin(String ip, int port, String user, String psw) throws Exception {
		JSch jsch = new JSch();
		Session session = (port <= 0) ? jsch.getSession(user, ip) : jsch.getSession(user, ip, port);
		session.setPassword(psw);
		session.setConfig("StrictHostKeyChecking", "no");
		session.connect();
		return session;
	}
	
	public static ChannelSftp getChannelSftp(Session session) throws Exception {
		Channel channel = session.openChannel("sftp");
		channel.connect();
		ChannelSftp channelSftp = (ChannelSftp) channel;
		return channelSftp;
	}
	
	public static String[][] getFileInfo(ChannelSftp cs, String dirPath, String fileScope) throws Exception {
		cs.cd(dirPath);
		String[] scopes = fileScope.split(",");
		if (scopes.length == 1) return getFileInfoFromVector(cs.ls(scopes[0]));
		
		String[][][] result = new String[scopes.length][][];
		int totalNum = 0;
		for (int i=0; i<scopes.length; i++) {
			Vector<?> v = cs.ls(scopes[i]);
			result[i] = getFileInfoFromVector(v);
			totalNum += result[i].length;
		}
		String[][] allFileInfo = new String[totalNum][];
		int position = 0;
		for (int i=0; i<result.length; i++) {
			String[][] info = result[i];
			System.arraycopy(info, 0, allFileInfo, position, info.length);
			position += info.length;
		}
		return allFileInfo;
	}
	
	public static String[][] getFileInfoFromVector(Vector<?> v) {
		String[][] fileInfo = new String[v.size()][2];
		for (int i=0; i<v.size(); i++) {
			String[] info = v.get(i).toString().replaceAll("\\s+", ",").split(",");
			//[-rw-rw-r--, 1, car, car, 190931, Dec, 11, 08:59, event.log.2016-12-11.08]
			fileInfo[i][0] = info[info.length-1];
			fileInfo[i][1] = info[4];
		}
		return fileInfo;
	}
	
	public static void downloadFile(ChannelSftp cs, String fileName, int fileSize, ByteBuffer buffer) throws Exception {
		InputStream is = cs.get(fileName);
		ReadableByteChannel rbc = new ReadableByteChannelRemote(is, fileSize);
		rbc.read(buffer);
		buffer.flip();
		byte[] content = new byte[buffer.remaining()];
		buffer.get(content);
		container.put(new TempFile(fileName, content));
		buffer.clear();
		rbc.close();
		is.close();
	}
	
	private static class TempFile implements Comparable<TempFile> {
		private String name;
		private byte[] content;
		
		public TempFile(String name, byte[] content) {
			super();
			this.name = name;
			this.content = content;
		}

		public int compareTo(TempFile o) {
			return o.content.length - this.content.length;
		}

		public String getName() {
			return name;
		}
		public byte[] getContent() {
			return content;
		}
	}
	
	public static class Downloader implements Runnable {
		private static final int FILE_SIZE_THRESHOLD = 1024 * 1024 * 100;
		private String ip;
		private int port;
		private String user;
		private String psw;
		private String dirPath;
		private String fileScope;
		
		public Downloader(Map<String,String> params) {
			super();
			this.ip = params.get("ip");
			this.port = (params.get("port")==null) ? 0 : Integer.valueOf(params.get("port"));
			this.user = params.get("user");
			this.psw = params.get("psw");
			this.dirPath = params.get("dirPath");
			this.fileScope = params.get("fileScope");
		}
		
		public void run() {
			Session session = null;
			ChannelSftp cs = null;
			try {
				session = sshLogin(ip, port, user, psw);
				cs = getChannelSftp(session);
				String[][] fileInfo = getFileInfo(cs, dirPath, fileScope);
				int maxFileSize = 0;
				for (String[] info : fileInfo) {
					int size = Integer.parseInt(info[1]);
					maxFileSize = (size>maxFileSize) ? size : maxFileSize;
				}
				ByteBuffer buffer = (maxFileSize > FILE_SIZE_THRESHOLD) ?
						getDirectByteBuffer(maxFileSize) : ByteBuffer.allocate(maxFileSize);
				for (int i=0; i<fileInfo.length; i++) {
					downloadFile(cs, fileInfo[i][0], Integer.parseInt(fileInfo[i][1]), buffer);
					System.out.println(">>>>downloaded file :" + Arrays.toString(fileInfo[i]));
				}
				sendCompletedNotification(fileInfo.length);
			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				cs.disconnect();
				session.disconnect();
			}
		}
	}
	
	public static void setInitialFlag(int threadNum) {
		flag.set(threadNum);
	}
	
	public static void sendCompletedNotification(int fileCount) {
		flag.incrementAndGet();
		System.out.println(">>>>RemoteExtractor finished a downloader, " + fileCount + " files in total");
	}
	
	@Deprecated//空实现
	public static ByteBuffer getDirectByteBuffer(int capacity) {
		//池，子类覆盖clear方法，归还给池。通过清理线程将懒惰的buffer清除(可否利用已有的缓存类？)。
		return ByteBuffer.allocateDirect(capacity);
	}
	
	public static List<Map<String,String>> analyzeFile(TempFile file) {
//		System.out.println(">>>>analyze file: " + file.getName());
		long startTime = System.currentTimeMillis();
		String str = new String(file.getContent(), StandardCharsets.UTF_8);
		StringTokenizer st = new StringTokenizer(str, "\n");
		List<Map<String, String>> list = new ArrayList<Map<String, String>>();
		while (st.hasMoreTokens()) {
			String line = st.nextToken();
			int index = line.indexOf("{\"code\":\"CALL_PHONE\"");
			if (index >= 0) {
				String json = line.substring(index);
				list.add(WebUtil.createMapFromJson(json));
			}
		}
		long costTime = System.currentTimeMillis() - startTime;
		System.out.println(">>>>analyze completed: [" + file.getName() + "] cost " + costTime + "ms");
		return list;
	}
	
	@Deprecated//抽离分析逻辑，抽象analyzeFile方法继承Callable，动态返回类型。分离getRemoteParams方法。
	public static class Analyzer implements Callable<List<Map<String,String>>> {
		public List<Map<String, String>> call() throws Exception {
			List<Map<String,String>> list = new ArrayList<Map<String,String>>();
			while (flag.get()>0 || container.size()>0) {
				TempFile file = container.poll(100, TimeUnit.MILLISECONDS);
				if (file == null) continue;
				list.addAll(analyzeFile(file));
			}
			return list;
		}
	}
	
	/*
	 * 线程池方式多线程执行与多Thread.start的权衡，直接返回结果还是再增加一个队列以缓冲？
	 * 前者阻塞时间过长，后者的必要性需要视后续操作而定，比如需要马上将结果持久化，则可以从队列彼端取数据。
	 * 动态构造队列类型。大文件的问题，平均文件大小决定内存占用，再选择队列类型。
	 */
	public static List<Map<String,String>> start(String[][] remoteParams) throws Exception {
		System.out.println(">>>>RemoteExtractor running...");
//		List<Map<String,String>> paramList = GeneralUtil.createListMap(remoteParams);
		List<Map<String,String>> paramList = null;
		int threadNum = paramList.size();
		container = new ArrayBlockingQueue<TempFile>(threadNum);
		setInitialFlag(threadNum);
		for (Map<String, String> map : paramList) {
			new Thread(new Downloader(map)).start();
		}
		List<List<Map<String,String>>> result = ConcurrentUtil.poolInvokeAll(Analyzer.class, threadNum);
		List<Map<String, String>> fullResult = new ArrayList<Map<String, String>>();
		for (List<Map<String,String>> list : result) {
			fullResult.addAll(list);
		}
		System.out.println(">>>>RemoteExtractor finished");
		return fullResult;
	}
	
	public static String[][] getRemoteParams(String start, String end) {
		String[] dates = GeneralUtil.getAllDatesBetweenTwoDate(start, end);
		StringBuilder scope = new StringBuilder();
		for (String date : dates) {
			scope.append("event.log." + date + "*,");
		}
		scope.deleteCharAt(scope.length()-1);
		String scopeStr = scope.toString();
		String[][] params = new String[][] {
			{"ip", "port", "user", "psw", "dirPath", "fileScope"},
			{"103.37.146.143", null, "car", "Huoli_carusr_1221",
					"/data/car/car-tomcat/logs/cardriver", scopeStr},
//			{"58.83.130.76", null, "caruser", "2vm&d3QsR",
//						"/home/caruser/car-tomcat/logs/cardriver", scopeStr}
			{"192.168.10.76", null, "caruser", "2vm&d3QsR",
				"/home/caruser/car-tomcat/logs/cardriver", scopeStr}
		};
		return params;
	}
	
	/**
	 * 源自Channels类，增加数据源长度属性以代替available()。
	 */
	private static class ReadableByteChannelRemote
			extends AbstractInterruptibleChannel
			implements ReadableByteChannel {
		private InputStream in;
		private int totalLength;
		private static final int TRANSFER_SIZE = 8192;
		private Object readLock = new Object();
		
		ReadableByteChannelRemote(InputStream in, int totalLength) {
			this.in = in;
			this.totalLength = totalLength;
		}
		
		public int read(ByteBuffer dst) throws IOException {
			if (dst.remaining() < totalLength) throw new IOException("缓冲区长度不足");
			byte[] buf = new byte[Math.min(totalLength,TRANSFER_SIZE)];
			int totalRead = 0;
			int bytesRead = 0;
			synchronized (readLock) {
				while (totalRead < totalLength) {
					int bytesToRead = Math.min((totalLength-totalRead), TRANSFER_SIZE);
					try {
						begin();
						bytesRead = in.read(buf, 0, bytesToRead);
					} finally {
						end(bytesRead > 0);
					}
					if (bytesRead < 0) break;
					totalRead += bytesRead;
					dst.put(buf, 0, bytesRead);
				}
				if ((bytesRead<0) && (totalRead==0)) return -1;
				return totalRead;
			}
		}
		
		protected void implCloseChannel() throws IOException {
			in.close();
		}
	}

	public static void main(String[] args) throws Exception {
		String[] lastWeek = GeneralUtil.getTwoDateOfLastWeek(true);
		String[][] params = getRemoteParams(lastWeek[0], lastWeek[1]);
		List<Map<String,String>> fullResult = start(params);
		for (int i = 0; i < fullResult.size(); i++) {
			System.out.println((i+1) + ":" + fullResult.get(i));
		}
	}
	
}